package com.itheima.kafka.controller;

import com.alibaba.fastjson.JSON;
import com.itheima.kafka.pojos.Order;
import com.itheima.kafka.pojos.Temperature;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.core.KafkaAdmin;
import org.springframework.stereotype.Controller;
import org.springframework.web.bind.annotation.*;

import java.time.LocalDateTime;
import java.time.format.DateTimeFormatter;
import java.util.*;
import java.util.concurrent.TimeUnit;

/**
 * API测试控制器 - 提供统一的API测试页面和接口
 * 集成所有example的功能到一个统一的API接口下
 */
@Slf4j
@Controller
@RequestMapping("/api")
public class ApiTestController {

    @Autowired
    private KafkaTemplate<String, String> kafkaTemplate;

    // ==================== 测试页面 ====================

    /**
     * 主测试页面
     */
    @GetMapping("")
    public String testPage() {
        return "api-test";
    }

    // ==================== Kafka连接测试 ====================
    
    /**
     * 测试Kafka连接状态
     */
    @GetMapping("/kafka/status")
    @ResponseBody
    public Map<String, Object> testKafkaConnection() {
        Map<String, Object> result = new HashMap<>();
        
        try {
            // 尝试获取Kafka集群信息
            String bootstrapServers = kafkaTemplate.getProducerFactory().getConfigurationProperties()
                .get("bootstrap.servers").toString();
            
            // 尝试发送一个测试消息来验证连接
            String testTopic = "connection-test-" + System.currentTimeMillis();
            String testMessage = "Kafka连接测试消息 - " + LocalDateTime.now().format(DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss"));
            
            // 同步发送测试消息（更简单的方式）
            try {
                kafkaTemplate.send(testTopic, testMessage).get(3, TimeUnit.SECONDS);
                log.info("✅ Kafka连接测试成功");
                result.put("connected", true);
                result.put("message", "Kafka连接正常");
            } catch (Exception e) {
                log.error("❌ Kafka连接测试失败", e);
                result.put("connected", false);
                result.put("message", "Kafka连接超时或失败: " + e.getMessage());
            }
            
            result.put("bootstrapServers", bootstrapServers);
            result.put("timestamp", LocalDateTime.now().format(DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss")));
            
        } catch (Exception e) {
            log.error("Kafka连接测试异常", e);
            result.put("connected", false);
            result.put("message", "Kafka连接异常: " + e.getMessage());
            result.put("timestamp", LocalDateTime.now().format(DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss")));
        }
        
        return result;
    }

    // ==================== 简单消息 API ====================
    
    /**
     * 发送简单消息
     */
    @GetMapping("/message/send")
    @ResponseBody
    public Map<String, Object> sendMessage(@RequestParam(defaultValue = "Hello Kafka!") String message) {
        String topic = "simple-messages";
        log.info("📤 [API] 发送简单消息: {}", message);
        kafkaTemplate.send(topic, message);
        
        Map<String, Object> response = new HashMap<>();
        response.put("success", true);
        response.put("topic", topic);
        response.put("message", message);
        response.put("timestamp", LocalDateTime.now().format(DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss")));
        return response;
    }
    
    /**
     * 发送带Key的消息
     */
    @GetMapping("/message/send/with-key")
    @ResponseBody
    public Map<String, Object> sendMessageWithKey(
            @RequestParam String key,
            @RequestParam String message) {
        String topic = "simple-messages";
        log.info("📤 [API] 发送带Key消息 - Key: {}, Message: {}", key, message);
        kafkaTemplate.send(topic, key, message);
        
        Map<String, Object> response = new HashMap<>();
        response.put("success", true);
        response.put("topic", topic);
        response.put("key", key);
        response.put("message", message);
        return response;
    }
    
    /**
     * 批量发送消息
     */
    @GetMapping("/message/send/batch")
    @ResponseBody
    public Map<String, Object> sendBatchMessages(@RequestParam(defaultValue = "5") int count) {
        String topic = "simple-messages";
        log.info("📤 [API] 批量发送 {} 条消息", count);
        
        for (int i = 1; i <= count; i++) {
            String message = "批量消息-" + i + " (" + LocalDateTime.now().format(DateTimeFormatter.ofPattern("HH:mm:ss")) + ")";
            kafkaTemplate.send(topic, message);
        }
        
        Map<String, Object> response = new HashMap<>();
        response.put("success", true);
        response.put("count", count);
        return response;
    }

    // ==================== Streams单词计数 API ====================
    
    /**
     * 发送文本进行单词计数
     */
    @GetMapping("/streams/word-count")
    @ResponseBody
    public Map<String, Object> sendTextForWordCount(@RequestParam(defaultValue = "hello kafka") String text) {
        String topic = "word-count-input";
        log.info("📤 [API/Streams] 发送文本: {}", text);
        kafkaTemplate.send(topic, text);
        
        Map<String, Object> response = new HashMap<>();
        response.put("success", true);
        response.put("input", text);
        response.put("说明", "文本已发送，Stream正在统计单词");
        return response;
    }
    
    /**
     * 发送示例文本进行单词计数
     */
    @GetMapping("/streams/word-count/samples")
    @ResponseBody
    public Map<String, Object> sendSamplesForWordCount() {
        String topic = "word-count-input";
        String[] samples = {"hello kafka", "hello world", "kafka streams", "hello streams"};
        
        log.info("📤 [API/Streams] 发送示例文本");
        for (String text : samples) {
            kafkaTemplate.send(topic, text);
            try { Thread.sleep(300); } catch (InterruptedException e) {}
        }
        
        Map<String, Object> response = new HashMap<>();
        response.put("success", true);
        response.put("count", samples.length);
        return response;
    }

    // ==================== 订单处理 API ====================
    
    /**
     * 创建订单
     */
    @GetMapping("/order/create")
    @ResponseBody
    public Map<String, Object> createOrder(
            @RequestParam(defaultValue = "笔记本电脑") String productName,
            @RequestParam(defaultValue = "1") Integer quantity,
            @RequestParam(defaultValue = "5999.0") Double price) {
        
        String topic = "orders";
        Order order = new Order(
            "ORD-" + System.currentTimeMillis(),
            productName, quantity, price,
            "PENDING", System.currentTimeMillis()
        );
        
        String orderJson = JSON.toJSONString(order);
        log.info("📦 [API/Order] 创建订单: {}", order);
        kafkaTemplate.send(topic, order.getOrderId(), orderJson);
        
        Map<String, Object> response = new HashMap<>();
        response.put("success", true);
        response.put("order", order);
        return response;
    }
    
    /**
     * 批量创建订单
     */
    @GetMapping("/order/create/batch")
    @ResponseBody
    public Map<String, Object> createBatchOrders(@RequestParam(defaultValue = "3") int count) {
        String topic = "orders";
        String[] products = {"iPhone 15", "MacBook Pro", "iPad Air"};
        List<Order> orders = new ArrayList<>();
        
        log.info("📦 [API/Order] 批量创建 {} 个订单", count);
        for (int i = 0; i < count; i++) {
            Order order = new Order(
                "ORD-" + System.currentTimeMillis() + "-" + i,
                products[i % products.length],
                (int)(Math.random() * 5) + 1,
                Math.round((Math.random() * 10000 + 1000) * 100.0) / 100.0,
                "PENDING", System.currentTimeMillis()
            );
            
            kafkaTemplate.send(topic, order.getOrderId(), JSON.toJSONString(order));
            orders.add(order);
            try { Thread.sleep(300); } catch (InterruptedException e) {}
        }
        
        Map<String, Object> response = new HashMap<>();
        response.put("success", true);
        response.put("count", count);
        response.put("orders", orders);
        return response;
    }

    // ==================== 温度传感器 API ====================
    
    /**
     * 上报温度数据
     */
    @GetMapping("/temperature/send")
    @ResponseBody
    public Map<String, Object> reportTemperature(
            @RequestParam(defaultValue = "sensor-001") String sensorId,
            @RequestParam(required = false) Double temperature) {
        
        String topic = "temperature-data";
        
        if (temperature == null) {
            temperature = Math.round((Math.random() * 20 + 15) * 10.0) / 10.0;
        }
        
        Temperature tempData = new Temperature(sensorId, temperature, System.currentTimeMillis());
        String tempJson = JSON.toJSONString(tempData);
        
        log.info("🌡️ [API/Temperature] 上报数据: {}", tempData);
        kafkaTemplate.send(topic, sensorId, tempJson);
        
        Map<String, Object> response = new HashMap<>();
        response.put("success", true);
        response.put("data", tempData);
        return response;
    }
    
    /**
     * 模拟多个传感器数据
     */
    @GetMapping("/temperature/send/simulate")
    @ResponseBody
    public Map<String, Object> simulateTemperatureSensors(@RequestParam(defaultValue = "10") int count) {
        String topic = "temperature-data";
        String[] sensors = {"sensor-001", "sensor-002", "sensor-003"};
        List<Temperature> dataList = new ArrayList<>();
        
        log.info("🌡️ [API/Temperature] 模拟 {} 次数据上报", count);
        for (int i = 0; i < count; i++) {
            String sensorId = sensors[i % sensors.length];
            double temp = Math.round((20 + (Math.random() - 0.5) * 10 + (Math.random() < 0.1 ? 20 : 0)) * 10.0) / 10.0;
            
            Temperature tempData = new Temperature(sensorId, temp, System.currentTimeMillis());
            kafkaTemplate.send(topic, sensorId, JSON.toJSONString(tempData));
            dataList.add(tempData);
            
            try { Thread.sleep(300); } catch (InterruptedException e) {}
        }
        
        Map<String, Object> response = new HashMap<>();
        response.put("success", true);
        response.put("count", count);
        response.put("dataList", dataList);
        return response;
    }
    
    /**
     * 上报高温告警
     */
    @GetMapping("/temperature/send/alert")
    @ResponseBody
    public Map<String, Object> reportAlertTemperature(
            @RequestParam(defaultValue = "sensor-001") String sensorId) {
        
        String topic = "temperature-data";
        double temperature = Math.round((Math.random() * 10 + 40) * 10.0) / 10.0;
        
        Temperature tempData = new Temperature(sensorId, temperature, System.currentTimeMillis());
        String tempJson = JSON.toJSONString(tempData);
        
        log.warn("🌡️ ⚠️ [API/Temperature] 异常高温: {}", tempData);
        kafkaTemplate.send(topic, sensorId, tempJson);
        
        Map<String, Object> response = new HashMap<>();
        response.put("success", true);
        response.put("data", tempData);
        response.put("alert", true);
        return response;
    }
}
